package com.mydubbo.registry;

import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import com.mydubbo.core.MyDubboRefer;
import com.mydubbo.core.MyDubboReferStarter;
import com.mydubbo.core.MyDubboService;
import com.mydubbo.util.Log;

@SuppressWarnings("resource")
public class ZookeeperRegistry implements Registry {
	
	private static CuratorFramework client = CuratorFrameworkFactory.newClient(
//			"127.0.0.1"
			"10.167.202.188:2181"
			, new ExponentialBackoffRetry(1000, 3));
	public static final String ROOT = "/mydubbo";
	static {
		try {
			client.start();
			Stat stat = client.checkExists().forPath(ROOT);
			if(stat==null) {
				client.create().forPath(ROOT);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void register(MyDubboService myDubboService) {
		try {
			String clazzName = myDubboService.getClazzName();
			String address = myDubboService.getAddress();
			String path = ROOT + "/" + clazzName;
			String providerPath = path + "/providers";
			String addressPath = providerPath + "/" + address;
			Stat stat = client.checkExists().forPath(addressPath);
			if(stat!=null) {
				throw new RuntimeException(address + "重复注册" + clazzName + "服务！！");
			}
			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(addressPath);
			Log.log(address + "注册" + clazzName + "服务成功...");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void subscribe(MyDubboRefer myDubboRefer) {
		try {
			String className = myDubboRefer.getClassName();
			String clientIp = myDubboRefer.getClientIp();
			String providerPath = ROOT + "/" + className + "/providers";
			String consumerPath = ROOT + "/" + className + "/consumers";
			String clientPath = consumerPath + "/" + clientIp;
//			Stat stat = client.checkExists().forPath(providerPath);
//			if(stat==null) {
//				client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(providerPath);
//			}
			Stat stat = client.checkExists().forPath(clientPath);
			if(stat!=null) {
				throw new RuntimeException(clientIp + "重复订阅" + className + "服务！！");
			}
			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(clientPath);
			Log.log(clientIp + "订阅" + className + "服务成功...");
			PathChildrenCache pcc = new PathChildrenCache(client, providerPath, true);
			pcc.getListenable().addListener(new PathChildrenCacheListener() {
				@Override
				public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
					Log.log(clientIp + "订阅到事件:\n\t" + event.getType() + "\n\t" + event.getData() + "\n\t" + event.getInitialData());
					List<String> serverIps = client.getChildren().forPath(providerPath);
					MyDubboReferStarter starter = MyDubboReferStarter.getInstance();
					Log.log("服务与提供者的关系原为=》" + starter.getServiceMap());
					starter.refreshServiceMap(className, serverIps);
					Log.log("更新为=》" + starter.getServiceMap());
				}
			});
			pcc.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		PathChildrenCache pcc = new PathChildrenCache(client, "/mydubbo/com.mydubbo.demo.UserService/providers", true);
		pcc.getListenable().addListener(new PathChildrenCacheListener() {
			@Override
			public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
				System.out.println("订阅到事件:\n\t" + event.getType() + "\n\t" + event.getData() + "\n\t" + event.getInitialData());
			}
		});
//		pcc.start();
		Thread.sleep(1000 * 3600);
	}

}
